package com.yj.rocketmq.listener;

import com.yj.rocketmq.RocketMqConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * Mq消息监听
 *
 * @author Xie
 * @date 19-3-19
 */
@Slf4j
public abstract class AbstractMessageListener implements MessageListenerOrderly {

    @Value("${spring.application.name:null}")
    private String consumer;

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext context) {
        MessageExt msg = null;
        try {
            // 消息为空
            if (list.isEmpty()) {
                log.info("{}消息内容为空", RocketMqConstants.LOG_PREFIX);
                return ConsumeOrderlyStatus.SUCCESS;
            }
            msg = list.get(0);
            if (msg == null) {
                log.error("{} 接收到消息体为空，无需处理，直接返回", RocketMqConstants.LOG_PREFIX);
                return ConsumeOrderlyStatus.SUCCESS;
            }
            log.info("{},接收消息：topic={},tags={},msgId={}",
                    RocketMqConstants.LOG_PREFIX, msg.getTopic(), msg.getTags(), msg.getMsgId());

            /**
             * 检查是否发送给本服务的消息
             */
            if (!checkIfMyMsg(msg)) {
                log.error("收到不是发给我本服务的消息 topic={},tags={}", msg.getTopic(), msg.getTags());
                return ConsumeOrderlyStatus.SUCCESS;
            }
            // 处理tag，tag中用可能拼hisId，去掉后面hosId
            String[] tempTag = msg.getTags().split("&&");
            return ack(msg, msg.getTopic(), tempTag[0].trim().toUpperCase());
        } catch (Exception e) {
            e.printStackTrace();

            //未到最大重试次数
            if (!RocketMqConstants.RETRY_TIMES.equals(msg.getReconsumeTimes())) {
                log.error("{}已尝试{} 次, 消费失败,topic={},tags={},msgId={}",
                        RocketMqConstants.LOG_PREFIX, msg.getReconsumeTimes(), msg.getTopic(), msg.getTags(), msg.getMsgId());
                return ConsumeOrderlyStatus.SUCCESS;
            }

            //检查是否达到重试次数
            log.warn("{}尝试{} 次, 到最大重试次数,topic={},tags={},msgId={}",
                    RocketMqConstants.LOG_PREFIX, RocketMqConstants.RETRY_TIMES, msg.getTopic(), msg.getTags(), msg.getMsgId());

            return ConsumeOrderlyStatus.SUCCESS;
        }
    }

    /**
     * 检查消息是否发给我的消息
     *
     * @param msg
     * @return
     */
    protected boolean checkIfMyMsg(MessageExt msg) {
        return true;
    }

    /**
     * 手动确认
     *
     * @param message
     * @param topic
     * @param tag
     * @return
     * @throws UnsupportedEncodingException
     */
    protected abstract ConsumeOrderlyStatus ack(MessageExt message, String topic, String tag) throws UnsupportedEncodingException;
}